package com.xiguthings.xiniu.iot.postproducer.worker.producer;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xiguthings.xiniu.iot.common.DeviceData;
import com.xiguthings.xiniu.iot.common.DeviceDataBody;
import com.xiguthings.xiniu.iot.common.DeviceDataHead;
import com.xiguthings.xiniu.iot.common.PostData;
import com.xiguthings.xiniu.iot.common.entity.ProducerRule;
import com.xiguthings.xiniu.iot.etce.PostRuleEtcdClient;
import com.xiguthings.xiniu.iot.postproducer.worker.controller.PostProducerContrller;

@Component
public class HandleThreadPool {
	private final Logger LOGGER = LoggerFactory.getLogger(HandleThreadPool.class);
	@Autowired
	private PostRuleEtcdClient ruleEtcdClient;
	@Autowired
	private PostProducerContrller postProducerContrller;
	@Value("${post-producer.pool.size}")
	private int threadCount;
	private ExecutorService threadPool;

	/**
	 * 初始化
	 */
	@PostConstruct
	void init() {
		threadPool = Executors.newFixedThreadPool(threadCount);
	}

	/**
	 * 程序关闭,关闭线程池
	 */
	@PreDestroy
	void fin() {
		shutdown();
	}

	public void shutdown() {
		if (threadPool != null)
			threadPool.shutdown();
		try {
			if (!threadPool.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
				LOGGER.info("Timed out waiting for consumer threads to shut down, exiting uncleanly");
			}
		} catch (InterruptedException e) {
			LOGGER.info("Interrupted during shutdown, exiting uncleanly");
		}
	}

	/**
	 * 执行发送post
	 *
	 * @param deviceDataStr
	 * @return
	 */
	public void execute(String deviceDataStr) {
		Runnable runnable = new Runnable() {
			@Override
			public void run() {
				DeviceData deviceData = formatMessage(deviceDataStr);
				Object devieceId = deviceData.getHead().getDeviceId();
				Integer ts = deviceData.getHead().getTs();
				LOGGER.info("\n收到的消息: {},{}", devieceId, ts);
				List<ProducerRule> producerRules = ruleEtcdClient
						.getPostProducerRuleByProjectId(deviceData.getHead().getProjectId());
				if (producerRules == null || producerRules.size() == 0) {
					return;
				}
				for (ProducerRule postProducerRule : producerRules) {
					String method = postProducerRule.getMethod();
					if (StringUtils.isBlank(method) || !"http".equals(method)) {
						continue;
					}
					List<String> urls = postProducerRule.getUrls();
					for (String url : urls) {
						if (StringUtils.isBlank(url)) {
							continue;
						}
						PostData postData = new PostData();
						postData.setBatchSize(postProducerRule.getBatchSize());
						postData.setInterval(postProducerRule.getInterval());
						postData.setUrl(url);
						String tempId = null;
						String dataStr = getDataStrByTemplateId(tempId, deviceData);
						postData.setData(dataStr);
						postProducerContrller.proucerMessage(postData);
						LOGGER.info("\n发送了数据: {},{}", devieceId, ts);
					}
				}
			}
		};
		threadPool.execute(runnable);
	}

	/**
	 * 根据deviceData，生产需要post的data的字符串
	 *
	 * @param tempId
	 * @param deviceData
	 * @return
	 */
	private String getDataStrByTemplateId(String tempId, DeviceData deviceData) {
		String data = null;
		// TODO 根据模板生产data的内容
		// if (StringUtils.isBlank(tempId)) {
		// 使用默认模板
		JSONObject dataJson = new JSONObject();
		// dataJson.put("meta", deviceData.getHead());
		dataJson.put("tags", deviceData.getHead());
		dataJson.put("data", deviceData.getBody());
		data = dataJson.toJSONString();
		// } else {
		// data = "没有对应的模板";
		// }
		return data;
	}

	/**
	 * 将收到的字符串，转成java对象
	 *
	 * @param dataStr
	 * @return
	 */
	private DeviceData formatMessage(String dataStr) {
		try {
			String[] partArr = dataStr.split("\\|\\|");
			DeviceData deviceData = new DeviceData();
			deviceData.setHead(JSON.parseObject(partArr[0], DeviceDataHead.class));
			deviceData.setBody(JSON.parseObject(partArr[1], DeviceDataBody.class));
			return deviceData;
		} catch (Exception e) {
			LOGGER.error("接收到kafka的数据格式不正确。收到的是消息:{}", dataStr);
		}
		return null;
	}

}
